package com.rrd.hive.udf;

import com.rrd.hive.util.IPLocation;
import com.rrd.hive.util.Location;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.ql.exec.Description;
import org.apache.hadoop.hive.ql.exec.UDF;

import java.io.BufferedReader;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.Map;


@Description(
        name = "ip_data",
        value = "_FUNC_(mobile) - Get area string from ip",
        extended = "Example:\n> SELECT _FUNC_('1.0.32.0');\n  {\"lng\":\"113.264434\",\"isp\":\"中国电信\",\"land\":\"亚洲\",\"country\":\"中国\",\"area_id\":\"\",\"city\":\"广州市\",\"region_id\":\"44000\",\"area\":\"\",\"country_english\":\"China\",\"international_code\":\"+86\",\"country_id\":\"CN\",\"region\":\"广东省\",\"lat\":\"23.129162\",\"city_id\":\"440100\"} \n> SELECT _FUNC_('1.0.32.0')['city'];\n北京"
)
public class IPDataUDF extends UDF {
    public static String dataFile = "/data/dim/data_ip/data_ip.dat";
    private static byte[] data;

    static {
        try {
            readAreaFile();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    public static Configuration config() {
        System.setProperty("HADOOP_USER_NAME", "hdfs");
        Configuration conf = new Configuration();
        conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
        return conf;
    }

    public static void readAreaFile() throws IOException {
        Configuration conf = config();
        FileSystem fs = FileSystem.newInstance(URI.create(dataFile), conf);
        Path remotePath = new Path(dataFile);
        if (fs.exists(remotePath)) {
            FSDataInputStream in = fs.open(remotePath);
            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(in));
            try{
                byte[] b = new byte[1024];
                ByteArrayOutputStream out = new ByteArrayOutputStream();
                while(in.read(b) != -1){
                    out.write(b);
                }
                data = out.toByteArray();
            }finally {
                bufferedReader.close();
                in.close();
                fs.close();
            }
        } else {
            System.out.println("找不到指定的文件");
        }
    }

    public Map<String, String> evaluate(String ip) throws Exception {
        if (ip == null || ip.equals("")) {
            return null;
        }
        IPLocation ipLocation = new IPLocation(data);
        Location location = ipLocation.fetchIPLocation(ip);
        return location.toMap();
    }

}
